package com.shujia.flink.source

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.source.SourceFunction

object Demo2Source {
  def main(args: Array[String]): Unit = {

    /**
      * 自定义source
      * 实现SourceFuncation
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义source
    val ds1 = env.addSource(new MySource)

    println(ds1.parallelism)

    val sink = ds1.print()
    //单独指定某一个算子并行度
    sink.setParallelism(2)


    env.execute("source")
  }

}


//自定义source
class MySource extends SourceFunction[Int] {

  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
    //生成数据

    var i = 0
    while (true) {
      i += 1
      //将数据发生到下游, 后面的算作
      ctx.collect(i)

      Thread.sleep(2000)
    }
  }

  override def cancel(): Unit = {

  }
}
